import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
import java.util.regex.Pattern;

public class TradeDataAnalysis {

    // 数据验证正则表达式（6位数字日期）
    private static final Pattern DATE_PATTERN = Pattern.compile("^\\d{6}$");
    private static final long MAX_RECORDS = 200000;  // 最大读取记录数

    // ------------------------------ Mapper 类（数据清洗与验证） ------------------------------
    public static class TradeMapper extends Mapper<LongWritable, Text, Text, Text> {
        private final Text outputKey = new Text();  // 格式：YYYY-MM,国家代号
        private final Text outputValue = new Text();  // 格式：标签:金额
        private long recordCount = 0;  // 已处理记录计数器

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 达到最大记录数时跳过后续处理
            if (recordCount >= MAX_RECORDS) {
                return;
            }

            String[] fields = value.toString().split(",");
            if (fields.length < 8) {
                context.getCounter("DataError", "InsufficientFields").increment(1);
                return;
            }

            // 提取字段（索引从0开始）
            String rawDate = fields[0].trim();    // 第1列：时间
            String tag = fields[1].trim();        // 第2列：标签
            String country = fields[2].trim();    // 第3列：国家代号
            String amountStr = fields[7].trim();  // 第8列：贸易额

            // 验证非空
            if (rawDate.isEmpty() || tag.isEmpty() || country.isEmpty() || amountStr.isEmpty()) {
                context.getCounter("DataError", "EmptyField").increment(1);
                return;
            }

            // 验证日期格式（6位数字）
            if (!DATE_PATTERN.matcher(rawDate).matches()) {
                context.getCounter("DataError", "InvalidDateFormat").increment(1);
                return;
            }

            // 验证年份（1988-2020）和月份（01-12）
            try {
                int year = Integer.parseInt(rawDate.substring(0, 4));
                int month = Integer.parseInt(rawDate.substring(4, 6));
                if (year < 1988 || year > 2020 || month < 1 || month > 12) {
                    context.getCounter("DataError", "InvalidDateRange").increment(1);
                    return;
                }
                // 格式化为YYYY-MM
                String formattedDate = String.format("%d-%02d", year, month);
                
                // 验证标签（1或2）
                if (!"1".equals(tag) && !"2".equals(tag)) {
                    context.getCounter("DataError", "InvalidTag").increment(1);
                    return;
                }

                // 验证金额（转换为长整型）
                long amount = Long.parseLong(amountStr);
                
                // 输出键：YYYY-MM,国家代号；值：标签:金额
                outputKey.set(formattedDate + "," + country);
                outputValue.set(tag + ":" + amount);
                context.write(outputKey, outputValue);
                
                // 记录数+1
                recordCount++;
            } catch (NumberFormatException e) {
                context.getCounter("DataError", "InvalidNumber").increment(1);
            }
        }
    }

    // ------------------------------ Reducer 类（聚合统计） ------------------------------
    public static class TradeReducer extends Reducer<Text, Text, Text, Text> {
        private final Text resultValue = new Text();

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            long exportAmount = 0;  // 出口额（标签1）
            long importAmount = 0;  // 进口额（标签2）

            for (Text value : values) {
                String[] parts = value.toString().split(":");
                if (parts.length != 2) continue;

                String tag = parts[0];
                long amount = Long.parseLong(parts[1]);

                if ("1".equals(tag)) {
                    exportAmount += amount;
                } else if ("2".equals(tag)) {
                    importAmount += amount;
                }
            }

            // 输出格式：日期,国家,出口额,进口额
            resultValue.set(exportAmount + "," + importAmount);
            context.write(key, resultValue);
        }
    }

    // ------------------------------ 主函数（作业配置） ------------------------------
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        // 设置输出分隔符为逗号（关键修改）
        conf.set("mapreduce.output.textoutputformat.separator", ",");
        Job job = Job.getInstance(conf, "TradeDataAnalysis");
        job.setJarByClass(TradeDataAnalysis.class);

        // 设置Mapper和Reducer
        job.setMapperClass(TradeMapper.class);
        job.setReducerClass(TradeReducer.class);

        // 设置输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        // 设置输入输出路径（从命令行参数获取）
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 提交作业
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}